package cn.me.implement.threadpool.v2_0;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * <h3>线程池 2.0 版本</h3>
 * 它实现的功能如下：
 * <ul>
 *     <li>使用锁来保证 {@link #threadPool} 相关操作的 <strong>互斥性</strong></li>
 *     <li>提供给使用者一个构造方法，指定 <strong>核心线程数量</strong> 和 <strong>最大线程数量</strong></li>
 *     <li>
 *         将线程分为两类：
 *         <ul>
 *             <li><strong>核心线程</strong>：永不停止</li>
 *             <li><strong>临时线程</strong>：在一段时间内没有获取任务的情况下停止</li>
 *         </ul>
 *     </li>
 *     <li>使用一个 <strong>{@link Set}</strong> 存放所有线程</li>
 *     <li><strong>在新建线程时交给线程一个任务</strong></li>
 *     <li><strong>线程在提交任务时才创建</strong></li>
 *     <li><strong>{@link #submit(Runnable task)} 通过任务队列提交任务，线程通过任务队列领取任务</strong></li>
 *     <li><strong>任务队列使用 {@link ArrayBlockingQueue} 阻塞队列实现</strong></li>
 * </ul>
 */
public class ThreadPool2_0 {

    /**
     * 线程池中核心线程的最大数量
     */
    private final int corePoolSize;

    /**
     * 线程池中线程的最大数量
     */
    private final int maxPoolSize;

    /**
     * 存放线程的集合，使用 {@link Set} 是因为 {@link Set#remove(Object)} 性能更高
     */
    private final Set<Worker> threadPool = new HashSet<>();

    /**
     * 线程池的管程
     * <p>
     * 用于保证 <strong>获取线程池大小</strong>、<strong>将线程放入线程池</strong>、<strong>从线程池中移除线程</strong> 的互斥性
     */
    private final Object threadPoolMonitor = new Object();

    /**
     * 任务队列，大小为 3，这是为了测试方便而特意设置的小
     */
    private final BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(3);

    /**
     * <h3>核心线程执行的任务</h3>
     * {@link #getTask()} 方法会一直阻塞，直到有新任务
     */
    public final class CoreWorker extends Worker {

        public CoreWorker(Runnable initialTask, Set<Worker> threadPool) {
            super(initialTask, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        protected void onWorkerExit() {
            // 在目前的代码中，发现核心线程并不会退出，所以这个方法先不实现
        }
    }

    /**
     * <h3>临时线程执行的任务</h3>
     * {@link #getTask()} 方法会在阻塞一定时间后如果还没有任务，则会返回 {@code null}
     */
    public final class TempWorker extends Worker {

        public TempWorker(Runnable initialTask, Set<Worker> threadPool) {
            super(initialTask, threadPool);
        }

        @Override
        protected Runnable getTask() {
            try {
                return taskQueue.poll(3, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        protected void onWorkerExit() {
            removeWorkerFromThreadPool(this);
        }
    }

    /**
     * 构造一个线程池
     *
     * @param corePoolSize 线程池中核心线程的最大数量
     * @param maxPoolSize 线程池中线程的最大数量
     */
    public ThreadPool2_0(int corePoolSize, int maxPoolSize) {
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
    }

    /**
     * 提交任务
     *
     * @param task 待执行的任务
     */
    public void submit(Runnable task) {
        // 如果 线程数量 小于 最大核心线程数量，则新建一个 核心线程 执行任务，然后直接返回
        synchronized (threadPoolMonitor) {
            if (threadPool.size() < corePoolSize) {
                CoreWorker coreWorker = new CoreWorker(task, threadPool);
                coreWorker.start();
                return;
            }
        }

        // 如果能够放到任务队列中，则直接返回
        if (taskQueue.offer(task)) {
            return;
        }

        // 如果 线程数量 小于 最大线程数量，则新建一个 临时线程 执行任务
        synchronized (threadPoolMonitor) {
            if (threadPool.size() < maxPoolSize) {
                TempWorker tempWorker = new TempWorker(task, threadPool);
                tempWorker.start();
            }
        }
    }

    /**
     * 获取当前线程池中的线程数量
     *
     * @return 当前线程池中的线程数量
     */
    public int getCurrPoolSize() {
        synchronized (threadPoolMonitor) {
            return threadPool.size();
        }
    }

    /**
     * 从 {@link #threadPool} 中移除指定的 {@link Worker} 对象
     *
     * @param worker 待移除的 {@link Worker} 对象
     */
    private void removeWorkerFromThreadPool(Worker worker) {
        synchronized (threadPoolMonitor) {
            threadPool.remove(worker);
        }
    }
}
